/*
 * socket.c
 *
 * Copyright (C) 2011 - Dr.NP
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

/**
 * @version socket.c 1.0.0
 * @package libbspcore
 * @author Dr.NP <np@bsgroup.org>
 * @update 03/18/2011
 */

/**
 * Network handler
 *
 * === CHANGELOG ===
 * [03/18/2011] - Creation
 */

#include "bsp.h"

#include "bsp_socket.h"
#include "bsp_thread.h"
#include "bsp_utils.h"

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/uio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <errno.h>
#include <err.h>
#include <math.h>
#include <pthread.h>

// In fact, we don't support *BSD ...
#ifdef HAVE_MALLOC_H
#ifndef __OpenBSD__
#include <malloc.h>
#endif
#endif

// Some platform need this
#ifndef __need_IOV_MAX
#define __need_IOV_MAX
#endif

#ifndef IOV_MAX
#define IOV_MAX                                 1024
#endif

// Server list
struct server_t *server_list;
size_t server_list_size;
int server_total;

// Client list
struct client_t **client_list;
size_t client_list_size;
int client_total;

// Connector list
struct connector_t **connector_list;
size_t connector_list_size;
int connector_total;

// Free clients
struct client_t **free_client_list;
size_t free_client_list_size;
int free_client_total;

// Free connectors
struct connector_t **free_connector_list;
size_t free_connector_list_size;
int free_connector_total;

// Send buffer lock
pthread_mutex_t client_send_buffer_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t client_list_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t connector_send_buffer_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t connector_list_lock = PTHREAD_MUTEX_INITIALIZER;

int new_server_local(struct server_t *);
int new_server_tcp(struct server_t *, int);
int new_server_udp(struct server_t *, int);

int new_client(struct server_t *);
int free_client(struct client_t *);

int new_connector_local(struct connector_t *);
int new_connector_tcp(struct connector_t *, int);
int new_connector_udp(struct connector_t *, int);
int free_connector(struct connector_t *);

int socket_init()
{
    // Pool initialization
    server_list_size = 
        server_total = 
        client_list_size = 
        client_total = 
        connector_list_size = 
        connector_total = 
        free_client_list_size = 
        free_client_total = 
        free_connector_list_size = 
        free_connector_total = 0;
    
    server_list = ensure_list_space(NULL, sizeof(struct server_t), &server_list_size, SERVER_LIST_INITIAL);
    client_list = ensure_list_space(NULL, sizeof(struct client_t *), &client_list_size, CLIENT_LIST_INITIAL);
    connector_list = ensure_list_space(NULL, sizeof(struct connector_t *), &connector_list_size, CONNECTOR_LIST_INITIAL);
    free_client_list = ensure_list_space(NULL, sizeof(struct client_t *), &free_client_list_size, CLIENT_LIST_INITIAL);
    free_connector_list = ensure_list_space(NULL, sizeof(struct connector_t *), &free_connector_list_size, CONNECTOR_LIST_INITIAL);

    if (!server_list || !client_list || !connector_list || !free_client_list || !free_connector_list)
    {
        err(RTN_ERROR_MEMORY_ALLOC, "Pool list initialize error!!!");
    }

    return RTN_SUCCESS;
}

// ################ SERVER ################

int new_server(
               int transport_type, 
               const char *listen_addr, 
               u_int16_t port)
{
    int fd = 0, i;
    int transport, epoll_mode, net;

    struct server_t *curr = NULL;

    // Why you create so many socket servers in a single process? so just loop list to find an empty slot...
    while (server_total >= server_list_size)
    {
        // What ...
        server_list = ensure_list_space(server_list, sizeof(struct server_t), &server_list_size, server_list_size * 2);
    }
    
    for (i = 0; i < server_list_size; i ++)
    {
        if (!server_list[i].in_use)
        {
            curr = &server_list[i];
            curr->id = i;
            
            break;
        }
    }

    if (!curr)
    {
        err(RTN_ERROR_MEMORY, "Server list pool error");
    }

    // Transport category : Unix Local / TCP / UDP
    if (transport_type & TRANSPORT_LOCAL)
    {
        transport = TRANSPORT_LOCAL;
    }

    else if (transport_type & TRANSPORT_UDP)
    {
        transport = TRANSPORT_UDP;
    }

    else
    {
        transport = TRANSPORT_TCP;
    }

    // Epoll mode : LT / ET
    if (transport_type & EPOLL_MODE_LT)
    {
        epoll_mode = EPOLL_MODE_LT;
    }

    else
    {
        epoll_mode = EPOLL_MODE_ET;
    }

    // IP type : IPv4 / IPv6
    if (transport_type & NET_IPV6)
    {
        net = NET_IPV6;
    }

    else
    {
        net = NET_IPV4;
    }
            
    switch (transport)
    {
        case TRANSPORT_LOCAL : 
            strncpy(curr->local_addr.sun_path, listen_addr, sizeof(curr->local_addr.sun_path) - 1);
            fd = new_server_local(curr);
            
            break;

        case TRANSPORT_TCP : 
            if (net == NET_IPV6)
            {
                inet_pton(AF_INET6, listen_addr, &curr->listen_addr6.sin6_addr);
                curr->listen_addr6.sin6_port = htons(port);
            }

            else
            {
                curr->listen_addr.sin_addr.s_addr = inet_addr(listen_addr);
                curr->listen_addr.sin_port = htons(port);
            }

            fd = new_server_tcp(curr, net);

            break;

        case TRANSPORT_UDP : 
            if (net == NET_IPV6)
            {
                inet_pton(AF_INET6, listen_addr, &curr->listen_addr6.sin6_addr);
                curr->listen_addr6.sin6_port = htons(port);
            }

            else
            {
                curr->listen_addr.sin_addr.s_addr = inet_addr(listen_addr);
                curr->listen_addr.sin_port = htons(port);
            }

            fd = new_server_udp(curr, net);

            break;
    }

    if (fd < 0)
    {
        debug_info("Socket server creation error\n");
        return RTN_ERROR_FATAL;
    }

    set_nonblock(curr->fd);

    int flag = 1;
    struct linger l = {.l_onoff = 1, .l_linger = 2};

    setsockopt(curr->fd, SOL_SOCKET, SO_KEEPALIVE, (void *) &flag, sizeof(flag));
    setsockopt(curr->fd, SOL_SOCKET, SO_LINGER, (void *) &l, sizeof(l));

    if (listen(curr->fd, SOCKET_LISTEN_QUEUES) < 0)
    {
        debug_info("Socket server listen error\n");
        return RTN_ERROR_SOCKET_LISTEN;
    }

    // Set epoll
    if (epoll_mode == EPOLL_MODE_LT)
    {
        curr->ev.events = EPOLLIN;
    }

    else
    {
        curr->ev.events = EPOLLIN | EPOLLET;
    }

    curr->ev.data.fd = curr->fd;
    set_fd(curr->fd, curr->id, FD_TYPE_SERVER);

    snprintf(curr->srv_addr, ADDR_DSP_LENGTH - 1, "%s", listen_addr);
    curr->srv_port = port;
    curr->conn_type = transport;
    curr->epoll_mode = epoll_mode;
    curr->net_type = net;
    curr->on_data = NULL;
    curr->on_conn = NULL;
    curr->on_close = NULL;
    curr->clients = 0;
    curr->bytes_read = curr->bytes_sent = 0;
    curr->srv_addr[0] = 0;
    
    curr->in_use = 0x1;
    server_total ++;

    debug_info("New %s server created at address %s port %d\n", 
               (transport == TRANSPORT_LOCAL) ? "Unix local IPC" : ((transport == TRANSPORT_UDP) ? "UDP" : "TCP"), 
               listen_addr, 
               port);
    
    return curr->id;
}

int new_server_local(struct server_t *s)
{
    struct timeval timeout = {.tv_sec = 3, .tv_usec = 0};
    int flag = 1;

    s->local_addr.sun_family = AF_UNIX;
    s->fd = socket(AF_UNIX, SOCK_STREAM, 0);

    if (!s->fd)
    {
        debug_info("Local socket create error\n");
        return RTN_ERROR_SOCKET_CREATE;
    }

    if (bind(s->fd, (struct sockaddr *) &s->local_addr, sizeof(&s->local_addr)) < 0)
    {
        debug_info("Local socket bind error\n");
        return RTN_ERROR_SOCKET_BIND;
    }

    setsockopt(s->fd, SOL_SOCKET, SO_SNDTIMEO, (void *) &timeout, sizeof(struct timeval));
    setsockopt(s->fd, SOL_SOCKET, SO_RCVTIMEO, (void *) &timeout, sizeof(struct timeval));
    setsockopt(s->fd, IPPROTO_TCP, TCP_NODELAY, (void *) &flag, sizeof(flag));
    
    return s->fd;
}

int new_server_tcp(struct server_t *s, int ip_type)
{
    int flag = 1;
    
    if (ip_type == NET_IPV6)
    {
        s->listen_addr6.sin6_family = AF_INET6;
        s->fd = socket(AF_INET6, SOCK_STREAM, 0);

        if (!s->fd)
        {
            debug_info("TCP socket create error\n");
            return RTN_ERROR_SOCKET_CREATE;
        }

        setsockopt(s->fd, SOL_SOCKET, SO_REUSEADDR, (void *) &flag, sizeof(flag));

        if (bind(s->fd, (struct sockaddr *) &s->listen_addr6, sizeof(s->listen_addr6)) < 0)
        {
            debug_info("TCP socket bind error\n");
            return RTN_ERROR_SOCKET_BIND;
        }
    }

    else
    {
        s->listen_addr.sin_family = AF_INET;
        s->fd = socket(AF_INET, SOCK_STREAM, 0);

        if (!s->fd)
        {
            debug_info("TCP socket create error\n");
            return RTN_ERROR_SOCKET_CREATE;
        }

        setsockopt(s->fd, SOL_SOCKET, SO_REUSEADDR, (void *) &flag, sizeof(flag));

        if (bind(s->fd, (struct sockaddr *) &s->listen_addr, sizeof(s->listen_addr)) < 0)
        {
            debug_info("TCP socket bind error\n");
            return RTN_ERROR_SOCKET_BIND;
        }
    }
    
    return s->fd;
}

int new_server_udp(struct server_t *s, int ip_type)
{
    int flag = 1;
    int snd_buf = SOCKET_SNDBUF;
    int rcv_buf = SOCKET_RCVBUF;
    
    if (ip_type == NET_IPV6)
    {
        s->listen_addr6.sin6_family = AF_INET6;
        s->fd = socket(AF_INET6, SOCK_DGRAM, 0);

        if (!s->fd)
        {
            debug_info("UDP socket create error\n");
            return RTN_ERROR_SOCKET_CREATE;
        }

        setsockopt(s->fd, SOL_SOCKET, SO_REUSEADDR, (void *) &flag, sizeof(flag));
        setsockopt(s->fd, SOL_SOCKET, SO_SNDBUF, (void *) &snd_buf, sizeof(snd_buf));
        setsockopt(s->fd, SOL_SOCKET, SO_RCVBUF, (void *) &rcv_buf, sizeof(rcv_buf));

        if (bind(s->fd, (struct sockaddr *) &s->listen_addr6, sizeof(s->listen_addr6)) < 0)
        {
            debug_info("UDP socket bind error\n");
            return RTN_ERROR_SOCKET_BIND;
        }
    }

    else
    {
        s->listen_addr.sin_family = AF_INET;
        s->fd = socket(AF_INET, SOCK_DGRAM, 0);

        if (!s->fd)
        {
            debug_info("UDP socket create error\n");
            return RTN_ERROR_SOCKET_CREATE;
        }

        setsockopt(s->fd, SOL_SOCKET, SO_REUSEADDR, (void *) &flag, sizeof(flag));
        setsockopt(s->fd, SOL_SOCKET, SO_SNDBUF, (void *) &snd_buf, sizeof(snd_buf));
        setsockopt(s->fd, SOL_SOCKET, SO_RCVBUF, (void *) &rcv_buf, sizeof(rcv_buf));

        if (bind(s->fd, (struct sockaddr *) &s->listen_addr, sizeof(s->listen_addr)) < 0)
        {
            debug_info("UDP socket bind error\n");
            return RTN_ERROR_SOCKET_BIND;
        }
    }
    
    return s->fd;
}

struct server_t * get_server(int srv_id)
{
    if (srv_id >= server_total || server_list[srv_id].in_use != 1)
    {
        return NULL;
    }

    return &server_list[srv_id];
}

// ################ CLIENT ################

int accept_client(struct server_t *s)
{
    socklen_t len = sizeof(struct sockaddr);

    if (!s || s->in_use != 1)
    {
        return RTN_ERROR_FATAL;
    }

    int client_id = new_client(s);

    if (client_id < 0)
    {
        return RTN_ERROR_GENERAL;
    }

    struct client_t *c = client_list[client_id];

    if (s->conn_type == TRANSPORT_LOCAL)
    {
        c->fd = accept(s->fd, (struct sockaddr *) &c->local_addr, &len);
    }

    else
    {
        if (s->net_type == NET_IPV4)
        {
            c->fd = accept(s->fd, (struct sockaddr *) &c->client_addr, &len);
        }

        else
        {
            c->fd = accept(s->fd, (struct sockaddr *) &c->client_addr6, &len);
        }
    }

    if (c->fd < 0)
    {
        free_client(c);

        return RTN_ERROR_SOCKET_ACCEPT;
    }

    set_nonblock(c->fd);

    c->srv_id = s->id;
    c->ev.data.fd = c->fd;

    if (c->epoll_mode == EPOLL_MODE_LT)
    {
        c->ev.events = EPOLLIN;
    }

    else
    {
        c->ev.events = EPOLLIN | EPOLLET;
    }

    set_fd(c->fd, c->id, FD_TYPE_CLIENT);
    dispatch_client(c->id);

    if (s->on_conn)
    {
        s->on_conn(client_id);
    }

    //debug_info("Client %d accepted by server %d\n", client_id, s->id);
    
    return client_id;
}

int close_client(int client_id)
{
    struct client_t *c = get_client(client_id);

    return free_client(c);
}

int new_client(struct server_t *s)
{
    struct client_t *curr = NULL;
    if (!s)
    {
        return RTN_ERROR_GENERAL;
    }

    pthread_mutex_lock(&client_list_lock);

    if (free_client_total > 0)
    {
        curr = free_client_list[-- free_client_total];
        client_list[curr->id] = curr;
    }

    else
    {
        curr = malloc(sizeof(struct client_t));
        if (!curr)
        {
            return RTN_ERROR_MEMORY_ALLOC;
        }

        memset(curr, 0, sizeof(struct client_t));

        while (client_total >= client_list_size)
        {
            client_list = ensure_list_space(client_list, sizeof(struct client_t *), &client_list_size, client_list_size * 2);
            if (!client_list)
            {
                err(RTN_ERROR_MEMORY_ALLOC, "Client list error!!!");
            }
        }

        client_list[client_total] = curr;
        curr->id = client_total;
    }

    pthread_mutex_unlock(&client_list_lock);

    curr->srv_id = s->id;
    curr->conn_type = s->conn_type;
    curr->epoll_mode = s->epoll_mode;
    curr->net_type = s->net_type;
    curr->read_buffer_data_cur = 0;

    curr->in_use = 1;
    curr->worker_id = 0;
    client_total ++;
    s->clients ++;

    return curr->id;
}

struct client_t * get_client(int client_id)
{
    struct client_t *c = NULL;

    if (client_id >= 0 && client_id < client_list_size)
    {
        c = client_list[client_id];
    }

    return c;
}

int free_client(struct client_t *c)
{
    struct server_t *s;

    if (!c)
    {
        return RTN_ERROR_GENERAL;
    }

    int id = c->id;
    int read_buffer_size = c->read_buffer_size;
    char *tmp_rbuf = c->read_buffer;
    int send_buffer_size = c->send_buffer_size;
    char *tmp_sbuf = c->send_buffer;

    pthread_mutex_lock(&client_list_lock);

    s = get_server(c->srv_id);

    if (s)
    {
        if (s->on_close && c->fd >= 0)
        {
            s->on_close(id);
        }

        s->clients --;
    }

    close(c->fd);

    memset(c, 0, sizeof(struct client_t));

    c->id = id;
    c->read_buffer_size = read_buffer_size;
    c->read_buffer = tmp_rbuf;
    c->send_buffer_size = send_buffer_size;
    c->send_buffer = tmp_sbuf;

    if (tmp_rbuf && read_buffer_size >= CLIENT_BUFFER_HIGHWAT)
    {
        free(c->read_buffer);
        c->read_buffer_size = 0;
    }

    if (tmp_sbuf && send_buffer_size >= CLIENT_BUFFER_HIGHWAT)
    {
        free(c->send_buffer);
        c->send_buffer_size = 0;
    }

    while (free_client_total >= free_client_list_size)
    {
        free_client_list = ensure_list_space(free_client_list, sizeof(struct client_t *), &free_client_list_size, free_client_list_size * 2);

        if (!free_client_list)
        {
            err(RTN_ERROR_MEMORY_ALLOC, "Free client list realloc error!!!");
        }
    }

    free_client_list[free_client_total ++] = c;
    client_total --;
    client_list[id] = NULL;

    pthread_mutex_unlock(&client_list_lock);

    return RTN_SUCCESS;
}

int append_client(int clt_id, const char *data, size_t len)
{
    struct client_t *c = get_client(clt_id);
    int tsize;
    char *tmp;

    if (!c || !data)
    {
        return RTN_ERROR_GENERAL;
    }

    if ((int) len <= 0)
    {
        len = strlen(data);
    }

    pthread_mutex_lock(&client_send_buffer_lock);

    if (!c->send_buffer)
    {
        if (len < CLIENT_BUFFER_INITIAL)
        {
            tsize = CLIENT_BUFFER_INITIAL;
        }

        else
        {
            tsize = 1 << (int) (log2(len - 1) + 1);
        }

        c->send_buffer = ensure_list_space(NULL, sizeof(char), &c->send_buffer_size, (size_t) tsize);
        if (!c->send_buffer)
        {
            pthread_mutex_unlock(&client_send_buffer_lock);
            
            return RTN_ERROR_MEMORY_ALLOC;
        }

        c->send_buffer_data_cur = c->send_buffer_data_size = 0;
    }

    while (c->send_buffer_size - c->send_buffer_data_size < len)
    {
        tmp = c->send_buffer;
        c->send_buffer = ensure_list_space(c->send_buffer, sizeof(char), &c->send_buffer_size, c->send_buffer_size * 2);
        if (!c->send_buffer)
        {
            c->send_buffer = tmp;
            pthread_mutex_unlock(&client_send_buffer_lock);

            return RTN_ERROR_MEMORY_ALLOC;
        }
    }

    memcpy(c->send_buffer + c->send_buffer_data_size, data, len);
    c->send_buffer_data_size += len;

    struct worker_t *w = get_worker(c->worker_id);
    if (w)
    {
        c->ev.events |= EPOLLOUT;
        epoll_ctl(w->epoll_fd, EPOLL_CTL_MOD, c->fd, &c->ev);
    }

    pthread_mutex_unlock(&client_send_buffer_lock);

    // First try send
    touch_worker(c->worker_id);

    return len;
}

int read_client(int clt_id)
{
    struct client_t *c = get_client(clt_id);
    struct server_t *s = NULL;
    int rlen, pdata, rread;
    char *tmp_buf, bad_read = 0x0;

    if (!c)
    {
        return RTN_ERROR_GENERAL;
    }

    if (!c->read_buffer)
    {
        c->read_buffer = ensure_list_space(NULL, sizeof(char), &c->read_buffer_size, (size_t) CLIENT_BUFFER_INITIAL);
        if (!c->read_buffer)
        {
            return RTN_ERROR_MEMORY_ALLOC;
        }

        c->read_buffer_data_cur = c->read_buffer_data_size = 0;
    }

    while (c->read_buffer_size <= c->read_buffer_data_size)
    {
        // Buffer full
        tmp_buf = c->read_buffer;
        c->read_buffer = ensure_list_space(c->read_buffer, sizeof(char), &c->read_buffer_size, c->read_buffer_size * 2);
        if (!c->read_buffer)
        {
            // Enlarge failed, just drop data and break out
            free(tmp_buf);

            return RTN_ERROR_FATAL;
        }
    }

    rread = 0;
    while (1)
    {
        rlen = recv(c->fd, 
                    c->read_buffer + c->read_buffer_data_size, 
                    c->read_buffer_size - c->read_buffer_data_size, 
                    0);
        
        if (rlen < 0)
        {
            bad_read = 0x1;
            break;
        }

        if (rlen == 0)
        {
            break;
        }

        else if (errno == EAGAIN || rlen < (c->read_buffer_size - c->read_buffer_data_size))
        {
            c->read_buffer_data_size += rlen;
            rread += rlen;

            break;
        }

        else
        {
            tmp_buf = c->read_buffer;
            c->read_buffer = ensure_list_space(c->read_buffer, sizeof(char), &c->read_buffer_size, c->read_buffer_size * 2);
            if (!c->read_buffer)
            {
                bad_read = 0x1;
                free(tmp_buf);
                
                break;
            }

            c->read_buffer_data_size += rlen;
            rread += rlen;
        }
    }

    if (0 == rlen || 0 == rread || bad_read)
    {
        free_client(c);

        return 0;
    }

    else
    {
        s = get_server(c->srv_id);

        if (s && (c->read_buffer_data_size - c->read_buffer_data_cur) && s->on_data)
        {
            pdata = s->on_data(c->id, 
                               c->read_buffer + c->read_buffer_data_cur, 
                               c->read_buffer_data_size - c->read_buffer_data_cur);
            c->read_buffer_data_cur += pdata;

            if (c->read_buffer_data_cur == c->read_buffer_data_size)
            {
                c->read_buffer_data_cur = c->read_buffer_data_size = 0;
            }
        }

        else
        {
            // Just drop data
            c->read_buffer_data_cur = c->read_buffer_data_size = 0;
        }
    }
    
    return rread;
}

int send_client(int clt_id)
{
    struct client_t *c = get_client(clt_id);
    int wlen = 0;

    if (!c)
    {
        return RTN_ERROR_GENERAL;
    }

    struct server_t *s = get_server(c->srv_id);

    //pthread_mutex_lock(&client_send_buffer_lock);

    if (c->send_buffer_data_cur < c->send_buffer_data_size)
    {
        wlen = send(c->fd, 
                    c->send_buffer + c->send_buffer_data_cur, 
                    c->send_buffer_data_size - c->send_buffer_data_cur, 
                    0);
    }

    c->send_buffer_data_cur += wlen;

    if (c->send_buffer_data_cur == c->send_buffer_data_size)
    {
        // All data sent, remove EPOLLOUT event from worker
        struct worker_t *w = get_worker(c->worker_id);
        if (w)
        {
            c->ev.events &= ~EPOLLOUT;
            epoll_ctl(w->epoll_fd, EPOLL_CTL_MOD, c->fd, &c->ev);
            c->send_buffer_data_cur = c->send_buffer_data_size = 0;
        }
    }

    //pthread_mutex_unlock(&client_send_buffer_lock);

    if (s)
    {
        s->bytes_sent += wlen;
    }
    
    return wlen;
}

// ################ CONNECTOR ################
int new_connector(
                  int transport_type, 
                  const char *remote_addr, 
                  u_int16_t port)
{
    int fd = 0;
    int transport, epoll_mode, net;
    
    struct connector_t *curr = NULL;

    pthread_mutex_lock(&connector_list_lock);

    if (free_connector_total > 0)
    {
        curr = free_connector_list[-- free_connector_total];
        connector_list[curr->id] = curr;
    }

    else
    {
        curr = malloc(sizeof(struct connector_t));
        if (!curr)
        {
            pthread_mutex_unlock(&connector_list_lock);
            return RTN_ERROR_MEMORY_ALLOC;
        }

        memset(curr, 0, sizeof(struct connector_t));
        
        while (connector_total >= connector_list_size)
        {
            connector_list = ensure_list_space(connector_list, sizeof(struct connector_t *), &connector_list_size, connector_list_size * 2);
            if (!connector_list)
            {
                err(RTN_ERROR_MEMORY_ALLOC, "Connector list error!!!");
            }
        }

        connector_list[connector_total] = curr;
        curr->id = connector_total;
    }

    pthread_mutex_unlock(&connector_list_lock);

    // Transport category : Unix Local / TCP / UDP
    if (transport_type & TRANSPORT_LOCAL)
    {
        transport = TRANSPORT_LOCAL;
    }

    else if (transport_type & TRANSPORT_UDP)
    {
        transport = TRANSPORT_UDP;
    }

    else
    {
        transport = TRANSPORT_TCP;
    }

    // Epoll mode : LT / ET
    if (transport_type & EPOLL_MODE_LT)
    {
        epoll_mode = EPOLL_MODE_LT;
    }

    else
    {
        epoll_mode = EPOLL_MODE_ET;
    }

    // IP type : IPv4 / IPv6
    if (transport_type & NET_IPV6)
    {
        net = NET_IPV6;
    }

    else
    {
        net = NET_IPV4;
    }

    switch (transport)
    {
        case TRANSPORT_LOCAL : 
            strncpy(curr->local_addr.sun_path, remote_addr, sizeof(curr->local_addr.sun_path) - 1);
            fd = new_connector_local(curr);
            
            break;

        case TRANSPORT_TCP : 
            if (net == NET_IPV6)
            {
                inet_pton(AF_INET6, remote_addr, &curr->remote_addr6.sin6_addr);
                curr->remote_addr6.sin6_port = htons(port);
            }

            else
            {
                curr->remote_addr.sin_addr.s_addr = inet_addr(remote_addr);
                curr->remote_addr.sin_port = htons(port);
            }

            fd = new_connector_tcp(curr, net);

            break;

        case TRANSPORT_UDP : 
            if (net == NET_IPV6)
            {
                inet_pton(AF_INET6, remote_addr, &curr->remote_addr6.sin6_addr);
                curr->remote_addr6.sin6_port = htons(port);
            }

            else
            {
                curr->remote_addr.sin_addr.s_addr = inet_addr(remote_addr);
                curr->remote_addr.sin_port = htons(port);
            }

            fd = new_connector_udp(curr, net);

            break;
    }

    if (fd < 0)
    {
        debug_info("Socket connector creation error\n");
        return RTN_ERROR_FATAL;
    }

    set_nonblock(curr->fd);

    int flag = 1;
    struct linger l = {.l_onoff = 1, .l_linger = 2};

    setsockopt(curr->fd, SOL_SOCKET, SO_KEEPALIVE, (void *) &flag, sizeof(flag));
    setsockopt(curr->fd, SOL_SOCKET, SO_LINGER, (void *) &l, sizeof(l));

    // Set epoll
    if (epoll_mode == EPOLL_MODE_LT)
    {
        curr->ev.events = EPOLLIN;
    }

    else
    {
        curr->ev.events = EPOLLIN | EPOLLET;
    }

    curr->ev.data.fd = curr->fd;
    set_fd(curr->fd, curr->id, FD_TYPE_CONNECTOR);    

    curr->conn_type = transport;
    curr->epoll_mode = epoll_mode;
    curr->net_type = net;

    curr->in_use = 1;
    curr->worker_id = 0;
    connector_total ++;

    return curr->id;
}

int new_connector_local(struct connector_t *n)
{
    struct timeval timeout = {.tv_sec = 3, .tv_usec = 0};
    int flag = 1;
    
    n->local_addr.sun_family = AF_UNIX;
    n->fd = socket(AF_UNIX, SOCK_STREAM, 0);

    if (!n->fd)
    {
        debug_info("Local socket create error\n");
        return RTN_ERROR_SOCKET_CREATE;
    }

    if (connect(n->fd, (struct sockaddr *) &n->local_addr, sizeof(n->local_addr)) < 0)
    {
        debug_info("Local socket connect error\n");
        return RTN_ERROR_SOCKET_CONNECT;
    }

    setsockopt(n->fd, SOL_SOCKET, SO_SNDTIMEO, (void *) &timeout, sizeof(struct timeval));
    setsockopt(n->fd, SOL_SOCKET, SO_RCVTIMEO, (void *) &timeout, sizeof(struct timeval));
    setsockopt(n->fd, IPPROTO_TCP, TCP_NODELAY, (void *) &flag, sizeof(flag));
    
    return n->fd;
}

int new_connector_tcp(struct connector_t *n, int ip_type)
{
    int flag = 1;
    
    if (ip_type == NET_IPV6)
    {
        n->remote_addr6.sin6_family = AF_INET6;
        n->fd = socket(AF_INET6, SOCK_STREAM, 0);

        if (!n->fd)
        {
            debug_info("TCP socket create error\n");
            return RTN_ERROR_SOCKET_CREATE;
        }

        setsockopt(n->fd, SOL_SOCKET, SO_REUSEADDR, (void *) &flag, sizeof(flag));

        if (connect(n->fd, (struct sockaddr *) &n->remote_addr6, sizeof(n->remote_addr6)) < 0)
        {
            debug_info("TCP socket bind error\n");
            return RTN_ERROR_SOCKET_CONNECT;
        }
    }

    else
    {
        n->remote_addr.sin_family = AF_INET;
        n->fd = socket(AF_INET, SOCK_STREAM, 0);

        if (!n->fd)
        {
            debug_info("TCP socket create error\n");
            return RTN_ERROR_SOCKET_CREATE;
        }

        setsockopt(n->fd, SOL_SOCKET, SO_REUSEADDR, (void *) &flag, sizeof(flag));

        if (connect(n->fd, (struct sockaddr *) &n->remote_addr, sizeof(n->remote_addr)) < 0)
        {
            debug_info("TCP socket bind error\n");
            return RTN_ERROR_SOCKET_CONNECT;
        }
    }
    
    return n->fd;
}

int new_connector_udp(struct connector_t *n, int ip_type)
{
    int flag = 1;
    int snd_buf = SOCKET_SNDBUF;
    int rcv_buf = SOCKET_RCVBUF;
    
    if (ip_type == NET_IPV6)
    {
        n->remote_addr6.sin6_family = AF_INET6;
        n->fd = socket(AF_INET6, SOCK_DGRAM, 0);

        if (!n->fd)
        {
            debug_info("UDP socket create error\n");
            return RTN_ERROR_SOCKET_CREATE;
        }

        setsockopt(n->fd, SOL_SOCKET, SO_REUSEADDR, (void *) &flag, sizeof(flag));
        setsockopt(n->fd, SOL_SOCKET, SO_SNDBUF, (void *) &snd_buf, sizeof(snd_buf));
        setsockopt(n->fd, SOL_SOCKET, SO_RCVBUF, (void *) &rcv_buf, sizeof(rcv_buf));

        if (connect(n->fd, (struct sockaddr *) &n->remote_addr6, sizeof(n->remote_addr6)) < 0)
        {
            debug_info("UDP socket bind error\n");
            return RTN_ERROR_SOCKET_CONNECT;
        }
    }

    else
    {
        n->remote_addr.sin_family = AF_INET;
        n->fd = socket(AF_INET, SOCK_DGRAM, 0);

        if (!n->fd)
        {
            debug_info("UDP socket create error\n");
            return RTN_ERROR_SOCKET_CREATE;
        }

        setsockopt(n->fd, SOL_SOCKET, SO_REUSEADDR, (void *) &flag, sizeof(flag));
        setsockopt(n->fd, SOL_SOCKET, SO_SNDBUF, (void *) &snd_buf, sizeof(snd_buf));
        setsockopt(n->fd, SOL_SOCKET, SO_RCVBUF, (void *) &rcv_buf, sizeof(rcv_buf));

        if (bind(n->fd, (struct sockaddr *) &n->remote_addr, sizeof(n->remote_addr)) < 0)
        {
            debug_info("UDP socket bind error\n");
            return RTN_ERROR_SOCKET_BIND;
        }
    }
    
    return n->fd;
}

struct connector_t * get_connector(int cnt_id)
{
    struct connector_t *n = NULL;

    if (cnt_id >= 0 && cnt_id < connector_total)
    {
        n = connector_list[cnt_id];
    }

    return n;
}

int free_connector(struct connector_t *n)
{
    if (!n)
    {
        return RTN_ERROR_GENERAL;
    }

    int id = n->id;
    int read_buffer_size = n->read_buffer_size;
    char *tmp_rbuf = n->read_buffer;
    int send_buffer_size = n->send_buffer_size;
    char *tmp_sbuf = n->send_buffer;

    pthread_mutex_lock(&connector_list_lock);

    if (n->on_close && n->fd >= 0)
    {
        n->on_close(id);
    }

    close(n->fd);

    memset(n, 0, sizeof(struct connector_t));

    n->id = id;
    n->read_buffer_size = read_buffer_size;
    n->read_buffer = tmp_rbuf;
    n->send_buffer_size = send_buffer_size;
    n->send_buffer = tmp_sbuf;

    if (tmp_rbuf && read_buffer_size >= CLIENT_BUFFER_HIGHWAT)
    {
        free(n->read_buffer);
        n->read_buffer_size = 0;
    }

    if (tmp_sbuf && send_buffer_size >= CLIENT_BUFFER_HIGHWAT)
    {
        free(n->send_buffer);
        n->send_buffer_size = 0;
    }

    while (free_connector_total >= free_connector_list_size)
    {
        free_connector_list = ensure_list_space(free_client_list, sizeof(struct connector_t *), &free_connector_list_size, free_connector_list_size * 2);

        if (!free_connector_list)
        {
            err(RTN_ERROR_MEMORY_ALLOC, "Free client list realloc error!!!");
        }
    }

    free_connector_list[free_connector_total ++] = n;
    connector_total --;
    connector_list[id] = NULL;

    pthread_mutex_unlock(&connector_list_lock);

    return RTN_SUCCESS;
}

int append_connector(int cnt_id, const char *data, size_t len)
{
    struct connector_t *n = get_connector(cnt_id);
    int tsize;
    char *tmp;

    if (!n || !data)
    {
        return RTN_ERROR_GENERAL;
    }

    if ((int) len <= 0)
    {
        len = strlen(data);
    }

    pthread_mutex_lock(&connector_send_buffer_lock);

    if (!n->send_buffer)
    {
        if (len < CLIENT_BUFFER_INITIAL)
        {
            tsize = CLIENT_BUFFER_INITIAL;
        }

        else
        {
            tsize = 1 << (int) (log2(len - 1) + 1);
        }

        n->send_buffer = ensure_list_space(NULL, sizeof(char), &n->send_buffer_size, (size_t) tsize);
        if (!n->send_buffer)
        {
            pthread_mutex_unlock(&connector_send_buffer_lock);
            
            return RTN_ERROR_MEMORY_ALLOC;
        }

        n->send_buffer_data_cur = n->send_buffer_data_size = 0;
    }

    while (n->send_buffer_size - n->send_buffer_data_size < len)
    {
        tmp = n->send_buffer;
        n->send_buffer = ensure_list_space(n->send_buffer, sizeof(char), &n->send_buffer_size, n->send_buffer_size * 2);
        if (!n->send_buffer)
        {
            n->send_buffer = tmp;
            pthread_mutex_unlock(&connector_send_buffer_lock);

            return RTN_ERROR_MEMORY_ALLOC;
        }
    }

    memcpy(n->send_buffer + n->send_buffer_data_size, data, len);
    n->send_buffer_data_size += len;

    struct worker_t *w = get_worker(n->worker_id);
    if (w)
    {
        n->ev.events |= EPOLLOUT;
        epoll_ctl(w->epoll_fd, EPOLL_CTL_MOD, n->fd, &n->ev);
    }

    pthread_mutex_unlock(&connector_send_buffer_lock);
    
    // First try send
    touch_worker(n->worker_id);
    
    return len;
}

int read_connector(int cnt_id)
{
    struct connector_t *n = get_connector(cnt_id);
    int rlen, pdata, rread;
    char *tmp_buf, bad_read = 0x0;

    if (!n)
    {
        return RTN_ERROR_GENERAL;
    }
    
    if (!n->read_buffer)
    {
        n->read_buffer = ensure_list_space(NULL, sizeof(char), &n->read_buffer_size, CLIENT_BUFFER_INITIAL);
        if (!n->read_buffer)
        {
            return RTN_ERROR_MEMORY_ALLOC;
        }

        n->read_buffer_data_cur = n->read_buffer_data_size = 0;
    }

    rread = 0;
    while (n->read_buffer_size <= n->read_buffer_data_size)
    {
        // Buffer full
        tmp_buf = n->read_buffer;
        n->read_buffer = ensure_list_space(n->read_buffer, sizeof(char), &n->read_buffer_size, n->read_buffer_size * 2);
        if (!n->read_buffer)
        {
            // Enlarge failed, just drop data and break out
            free(tmp_buf);

            return RTN_ERROR_FATAL;
        }
    }

    while (1)
    {
        rlen = recv(n->fd, 
                    n->read_buffer + n->read_buffer_data_size, 
                    n->read_buffer_size - n->read_buffer_data_size, 
                    0);

        if (rlen < 0)
        {
            bad_read = 0x1;
            break;
        }

        else if (rlen == 0)
        {
            break;
        }

        else if (errno == EAGAIN || rlen < (n->read_buffer_size - n->read_buffer_data_size))
        {
            n->read_buffer_data_size += rlen;
            rread += rlen;
            
            break;
        }

        else
        {
            tmp_buf = n->read_buffer;
            n->read_buffer = ensure_list_space(n->read_buffer, sizeof(char), &n->read_buffer_size, n->read_buffer_size * 2);
            if (!n->read_buffer)
            {
                bad_read = 0x1;
                free(tmp_buf);

                break;
            }

            n->read_buffer_data_size += rlen;
            rread += rlen;
        }
    }

    if (0 == rlen || 0 == rread || bad_read)
    {
        free_connector(n);

        return 0;
    }

    if ((n->read_buffer_data_size - n->read_buffer_data_cur) && n->on_data)
    {
        pdata = n->on_data(n->id, 
                           n->read_buffer + n->read_buffer_data_cur, 
                           n->read_buffer_data_size - n->read_buffer_data_cur);
        n->read_buffer_data_cur += pdata;

        if (n->read_buffer_data_cur == n->read_buffer_data_size)
        {
            // All data gone
            n->read_buffer_data_cur = n->read_buffer_data_size = 0;
        }
    }

    else
    {
        n->read_buffer_data_cur = n->read_buffer_data_size = 0;
    }
    
    return rread;
}

int send_connector(int cnt_id)
{
    struct connector_t *n = get_connector(cnt_id);
    int wlen = 0;

    if (!n)
    {
        return RTN_ERROR_GENERAL;
    }
/*
#ifdef BSP_NO_WORKER
    debug_hex_data(n->send_buffer + n->send_buffer_data_cur, (n->send_buffer_data_size - n->send_buffer_data_cur));
    
    while (n->send_buffer_data_cur < n->send_buffer_data_size)
    {
        wlen = send(n->fd, n->send_buffer + n->send_buffer_data_cur, (n->send_buffer_data_size - n->send_buffer_data_cur), 0);
        n->send_buffer_data_cur += wlen;
        n->bytes_sent += wlen;
    }
#else
*/
    
    if (n->send_buffer_data_cur < n->send_buffer_data_size)
    {
        wlen = send(n->fd, n->send_buffer + n->send_buffer_data_cur, (n->send_buffer_data_size - n->send_buffer_data_cur), 0);
    }

    n->send_buffer_data_cur += wlen;

    if (n->send_buffer_data_cur == n->send_buffer_data_size)
    {
        // All clear
        struct worker_t *w = get_worker(n->worker_id);
        if (w)
        {
            n->ev.events &= ~EPOLLOUT;
            epoll_ctl(w->epoll_fd, EPOLL_CTL_MOD, n->fd, &n->ev);
            n->send_buffer_data_cur = n->send_buffer_data_size = 0;
        }
    }

    n->bytes_sent += wlen;
    
    return wlen;
}
